package drds.global_transaction.api.rpc.rpc_server;

import drds.global_transaction.api.rpc.AbstractChannelDuplexHandler;
import drds.global_transaction.api.rpc.ByteAndRpcInfoCodec;
import drds.global_transaction.api.rpc.config.ServerConfig;
import drds.global_transaction.common.thread.NamedThreadFactory;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.concurrent.ThreadPoolExecutor;

@Slf4j
public abstract class AbstractRpcServer extends AbstractChannelDuplexHandler implements IRpcServer
{
	private final ServerBootstrap serverBootstrap;
	private final EventLoopGroup workerEventLoopGroup;
	private final EventLoopGroup bossEventLoopGroup;
	private final ServerConfig serverConfig;
	@Setter
	@Getter
	private DefaultEventExecutorGroup defaultEventExecutorGroup;
	@Setter
	@Getter
	private int listenPort;

	public AbstractRpcServer(final ServerConfig serverConfig, //
			final ThreadPoolExecutor threadPoolExecutor, //
			final ChannelHandler... channelHandlers)
	{//
		super(threadPoolExecutor);
		this.serverBootstrap = new ServerBootstrap();
		this.serverConfig = serverConfig;
		this.bossEventLoopGroup = new NioEventLoopGroup(serverConfig.getBossThreadSize(), new NamedThreadFactory(serverConfig.getBossThreadPrefix(), serverConfig.getBossThreadSize()));
		if (ServerConfig.enableEpoll())
		{
			this.workerEventLoopGroup = new EpollEventLoopGroup(serverConfig.getServerWorkerThreads(), new NamedThreadFactory(serverConfig.getWorkerThreadPrefix(), serverConfig.getServerWorkerThreads()));
		} else
		{
			this.workerEventLoopGroup = new NioEventLoopGroup(serverConfig.getServerWorkerThreads(), new NamedThreadFactory(serverConfig.getWorkerThreadPrefix(), serverConfig.getServerWorkerThreads()));
		}
		if (null != channelHandlers)
		{
			this.channelHandlers = channelHandlers;
		}
		// init listenPort in constructor so that getListenPort() will always get the
		// exact port
		setListenPort(serverConfig.getDefaultListenPort());
	}

	@Override
	public void returnChannel(String serverAddress, Channel channel)
	{
		log.info("will destroy channel:" + channel + ",address:" + serverAddress);
		channel.disconnect();
		channel.close();
	}

	@Override
	public void shutdown()
	{
		try
		{
			this.bossEventLoopGroup.shutdownGracefully();
			this.workerEventLoopGroup.shutdownGracefully();
			if (this.defaultEventExecutorGroup != null)
			{
				this.defaultEventExecutorGroup.shutdownGracefully();
			}
		} catch (Exception exx)
		{
			log.error(exx.getMessage());
		}
	}

	@Override
	public void start()
	{
		this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(serverConfig.getServerWorkerThreads(), new NamedThreadFactory(serverConfig.getExecutorThreadPrefix(), serverConfig.getServerWorkerThreads()));

		this.serverBootstrap.group(this.bossEventLoopGroup, this.workerEventLoopGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, serverConfig.getSoBackLogSize()).option(ChannelOption.SO_REUSEADDR, true).option(ChannelOption.SO_KEEPALIVE, true).childOption(ChannelOption.TCP_NODELAY, true).childOption(ChannelOption.SO_SNDBUF, serverConfig.getServerSocketSendBufSize()).childOption(ChannelOption.SO_RCVBUF, serverConfig.getServerSocketResvBufSize()).childOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, serverConfig.getWriteBufferHighWaterMark()).childOption(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, serverConfig.getWriteBufferLowWaterMark()).localAddress(new InetSocketAddress(listenPort)).childHandler(new ChannelInitializer<SocketChannel>()
		{
			@Override
			public void initChannel(SocketChannel socketChannel)
			{
				/**
				 * 服务端 入口
				 */
				socketChannel.pipeline().addLast(new IdleStateHandler(serverConfig.getChannelMaxReadIdleSeconds(), 0, 0))//
						.addLast(new ByteAndRpcInfoCodec());
				if (null != channelHandlers)
				{
					addChannelHandlersToChannelPipelineLast(socketChannel, channelHandlers);
				}

			}
		});

		if (serverConfig.isEnableServerPooledByteBufAllocator())
		{
			this.serverBootstrap.childOption(ChannelOption.ALLOCATOR, ServerConfig.DIRECT_BYTE_BUF_ALLOCATOR);
		}

		try
		{
			log.info("start bind port: " + listenPort);
			ChannelFuture channelFuture = this.serverBootstrap.bind(listenPort).sync();
			log.info("bind port: " + listenPort + " ok");
			channelFuture.channel().closeFuture().sync();
		} catch (InterruptedException exx)
		{
			throw new RuntimeException(exx);
		}

	}

}
